package com.bbx.flink.demo.ProcessFunctionAllDemo;

import com.bbx.flink.demo.entity.Student;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

@Slf4j
public class CoProcessFunctionDemo {

    public static void  main(String [] args ) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Student> source1 =env.fromElements(
                new Student("1", "s1"),
                new Student("2", "s2"),
                new Student("3", "s3"),
                new Student("4", "s4"),
                new Student("5", "s5"),
                new Student("6", "s6"),
                new Student("7", "s7"),
                new Student("8", "s8"),
                new Student("9", "s9"),
                new Student("10", "s10")
        );

        DataStreamSource<Student> source2 = env.fromElements(
                new Student("1", 11, "男"),
                new Student("2", 12, "男"),
                new Student("3", 13, "女"),
                new Student("4", 14, "男"),
                new Student("5", 15, "女"),
                new Student("6", 16, "男"),
                new Student("7", 17, "女"),
                new Student("8", 18, "男"),
                new Student("9", 19, "女"),
                new Student("10", 100, "女")
        );


        source1.connect(source2).process(new MyCoProcessFunction()).print();
        env.execute();
    }


}



@Slf4j
class MyCoProcessFunction  extends CoProcessFunction<Student,Student,Student>{
    private static final long serialVersionUID = -2115111985092377828L;

    private Map<String ,String>  map = new HashMap<>();


    @Override
    public void processElement1(Student value, Context ctx, Collector<Student> out) throws Exception {
        map.put(value.getId(),value.getName());

    }

    @Override
    public void processElement2(Student value, Context ctx, Collector<Student> out) throws Exception {
        String name = map.get(value.getId());
        value.setName(name);
        Long timestamp = ctx.timestamp();
        long currentProcessingTime = ctx.timerService().currentProcessingTime();
        log.info("~~~~~~~~~~~~~~~~~~~~~~~timestamp[{}]",timestamp);
        log.info("~~~~~~~~~~~~~~~~~~~~~~~currentProcessingTime[{}]",currentProcessingTime);


        out.collect(value);
    }
}